package com.teamsun.mqtt.service;

import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import io.netty.handler.codec.mqtt.MqttPublishVariableHeader;
import io.netty.handler.codec.mqtt.MqttQoS;

import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;


import org.apache.ibatis.session.SqlSession;
import org.fusesource.mqtt.client.QoS;

import com.teamsun.mqtt.db.DBSessionFactory;
import com.teamsun.mqtt.entity.MsgRep;
import com.teamsun.mqtt.entity.SendMsg;
import com.teamsun.mqtt.entity.vo.DelaySnedMsg;

/**
 * 
 * @author tzj
 *
 */
public class MessageDataService {

	static final   String  updateSendMsg="com.teamsun.mqtt.entity.SendMsgMapper.updateState";
	
	static final  String saveSendMsg="com.teamsun.mqtt.entity.SendMsgMapper.insert";
	
	
	public static final  int MSG_SENDSUCCESS=2;
	public static final  int MSG_SENDINDB=1;
	public static final  int MSG_SENDERROR=0;
	
	
	public static	final  int  MAXCACHEMSGSIZE=10;

	static ConcurrentHashMap<String, Channel> str2channel=ChannelDataService.str2channel;
	/**
	 * 每一个客户端订阅的主题
	 */
	static ConcurrentHashMap<String, BlockingQueue<String>> submap=new ConcurrentHashMap<String, BlockingQueue<String>>();
	/**
	 * 主题消息 不清除
	 */
	static ConcurrentHashMap<String, Integer> topContent=new ConcurrentHashMap<String, Integer>();

	/**
	 * 消息 队列
	 */
	static BlockingQueue<MsgRep> msgRepsque=new LinkedBlockingQueue<MsgRep>(MAXCACHEMSGSIZE);
	
	
	/**
	 * 已经发送过的消息
	 */
	static  BlockingQueue<SendMsg> msgSendedQue=new LinkedBlockingQueue<SendMsg>(MAXCACHEMSGSIZE) ;
	
	
	/**
	 * 准备更改状态的 已发送消息
	 */
	static  BlockingQueue<SendMsg> readUpdateSendedQue=new LinkedBlockingQueue<SendMsg>(MAXCACHEMSGSIZE) ;
	
	
	static  BlockingQueue<SendMsg> readySendedQue=new LinkedBlockingQueue<SendMsg>(MAXCACHEMSGSIZE) ;

	
	/**
	 * 已经发送过的消息  不能清除
	 */
	static ConcurrentHashMap<String, ConcurrentLinkedQueue<Integer>> messageSends=
			new ConcurrentHashMap<String, ConcurrentLinkedQueue<Integer>>();
	
	
	static ExecutorService dboptgroup=Executors.newFixedThreadPool(8);
	

	/**
	 * 对客户端发送数据
	 * 获取消息的主题 根据主题得到客户端  
	 * 把消息发送给这些客户端
	 * @param transportMessage
	 */
	public static void sendPubMsg(MsgRep transportMessage,MqttQoS qos) {

		
		BlockingQueue<String> idens=submap.get(transportMessage.getTopname());
		
		if(idens==null)
			return;
		 Iterator<String> iterator=idens.iterator();
		 
		 while(iterator.hasNext()){
			 String clientid=iterator.next();
		
			 saveStateInDb(transportMessage, qos, clientid);
			
			 sendMsg2Client(transportMessage,clientid,qos);

		}
	}
	
	
/**
 * 实际的发送操作
 * @param transportMessage
 * @param times
 * @param clientid
 * @return
 */
	public  static void  sendMsg2Client(MsgRep transportMessage,String clientid,MqttQoS  qos){
		
		 Channel channel = str2channel.get(clientid);

			if (channel != null) {
				
				MqttFixedHeader Header = new MqttFixedHeader(
						MqttMessageType.PUBLISH, true,
						MqttQoS.EXACTLY_ONCE, false, 0);
				MqttPublishVariableHeader publishVariableHeader =
					new MqttPublishVariableHeader(
						transportMessage.getTopname(),
						transportMessage.getId());

				MqttPublishMessage publishMessage = new MqttPublishMessage(
						Header, publishVariableHeader,
						Unpooled.copiedBuffer(transportMessage.getContent()));
				
				 channel.writeAndFlush(publishMessage);
				 
				 /**
				  * 放入重发队列 设置超时时间为一分钟
				  */
				
				 if(qos.equals(MqttQoS.EXACTLY_ONCE)){
					 Calendar calendar=Calendar.getInstance();
					 calendar.add(Calendar.MINUTE, 1);
					 DelaySnedMsg delaySnedMsg=new DelaySnedMsg(transportMessage,
							 clientid,calendar.getTime().getTime());
					 ReSendService.getIntanceo().submit(delaySnedMsg);
				 }
	
				System.out.println("对" + clientid + "发送了 id:"+transportMessage.getId()+"消息："
						+ new String(transportMessage.getContent(), Charset.forName("UTF-8")));
			}
	}
	
	/**
	 * 把对客户端发送消息的状态保存到数据库
	 * @param transportMessage 消息
	 * @param times 标识成功或者失败
	 * @param clientid 推送给客户端的客户端标识
	 */
	private  static  void  saveStateInDb(MsgRep transportMessage,MqttQoS qoS,String clientid){
		 SendMsg msg=null;
		 switch(qoS){
		 	case EXACTLY_ONCE:
		 		
		 		msg=new SendMsg(clientid, transportMessage.getId(),
		 				MSG_SENDINDB,  transportMessage.getSendiden());
		 		break;
		 	default:
		 		msg=new SendMsg(clientid, transportMessage.getId(),
		 				MSG_SENDSUCCESS,  transportMessage.getSendiden());
		 		
		 }
		 
		 if(msg!=null)
			 saveSendMsg(msg);
	}
	/**
	 *修改sendmsg状态
	 * @param sedmsgs
	 */
	private static  void  submitUpdateSendedMsg(final List<SendMsg> sedmsgs){
		
		dboptgroup.submit(new Runnable() {
			
			@Override
			public void run() {
				
				SqlSession session= DBSessionFactory.getSqlSession();
				for(SendMsg sendMsg:sedmsgs)	
					session.update(updateSendMsg,sendMsg);
					
				
				session.commit();
				session.close();
				
			}
		});
		
	}
	
	/**
	 * 更改发送日志的消息
	 * @param messageid
	 * @param iden
	 */
	public static void   updateSendMsg(String clientiden,Integer msgid){

		SendMsg msg=new SendMsg();
		msg.setSendmsgid(msgid);
		msg.setClientid(clientiden);
		msg.setSendtimes(MSG_SENDSUCCESS);
		if(!readUpdateSendedQue.offer(msg)){
			List<SendMsg> msgs=new ArrayList<SendMsg>();
			readUpdateSendedQue.drainTo(msgs);
			msgs.add(msg);
			submitUpdateSendedMsg(msgs);
			
		}
		
	}
	
	/**
	 * 保存发送消息
	 *如果消息是标识成功的进入msgSendedQue队列
	 *如果是标识 indb的，进入readySendedQue
	 * @param sendMsg
	 */
	private  static void   saveSendMsg(SendMsg sendMsg){

		
		if(sendMsg.getSendtimes()<2){
			if(!readySendedQue.offer(sendMsg)){
				List<SendMsg> msgs=new ArrayList<SendMsg>();
				readySendedQue.drainTo(msgs);
				msgs.add(sendMsg);
				submitSendedMsg(msgs);
				
			}
		}else{
			if(!msgSendedQue.offer(sendMsg)){
				List<SendMsg> msgs=new ArrayList<SendMsg>();
				msgSendedQue.drainTo(msgs);
				msgs.add(sendMsg);
				submitSendedMsg(msgs);
				
			}
		}
		
		
		
	}
	
	/**
	 * 提交sendmsg
	 * @param sedmsgs
	 */
	private static void  submitSendedMsg(final List<SendMsg> sedmsgs){
		
		dboptgroup.submit(new Runnable() {
			
			@Override
			public void run() {
				
				SqlSession session= DBSessionFactory.getSqlSession();
				for(SendMsg sendMsg:sedmsgs){					
					session.insert(saveSendMsg,sendMsg);
				}
				session.commit();
				session.close();
				
			}
		});
		
	}
	
	public static ConcurrentHashMap<String, Channel> getStr2channel() {
		return str2channel;
	}
	public static void setStr2channel(ConcurrentHashMap<String, Channel> str2channel) {
		MessageDataService.str2channel = str2channel;
	}
	
	
	
	public static ConcurrentHashMap<String, BlockingQueue<String>> getSubmap() {
		return submap;
	}

	public static void setSubmap(
			ConcurrentHashMap<String, BlockingQueue<String>> submap) {
		MessageDataService.submap = submap;
	}

	public static ConcurrentHashMap<String, Integer> getTopContent() {
		return topContent;
	}
	public static void setTopContent(ConcurrentHashMap<String, Integer> topContent) {
		MessageDataService.topContent = topContent;
	}
	
	public static ConcurrentHashMap<String, ConcurrentLinkedQueue<Integer>> getMessageSends() {
		return messageSends;
	}
	public static void setMessageSends(
			ConcurrentHashMap<String, ConcurrentLinkedQueue<Integer>> messageSends) {
		MessageDataService.messageSends = messageSends;
	}
	public static BlockingQueue<MsgRep> getMsgRepsque() {
		return msgRepsque;
	}
	public static void setMsgRepsque(BlockingQueue<MsgRep> msgRepsque) {
		MessageDataService.msgRepsque = msgRepsque;
	}
	


	public static BlockingQueue<SendMsg> getMsgSendedQue() {
		return msgSendedQue;
	}


	public static void setMsgSendedQue(BlockingQueue<SendMsg> msgSendedQue) {
		MessageDataService.msgSendedQue = msgSendedQue;
	}
	
	
	
	
}
